package com.atguigu.edu.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.DwsTrafficPageViewWindowBean;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsTrafficPageViewWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTrafficPageViewWindow().start(10023, 4, "dws_play_video_chapter_window", Constant.TOPIC_DWD_PLAY_VIDEO_DETAIL);

    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        //从kafka中读取的log数据中过滤出page的page_id为home、course_list、coure_detail的数据,并转化成实体类
        SingleOutputStreamOperator<DwsTrafficPageViewWindowBean> beanDS = kafkaStrDS.process(new ProcessFunction<String, DwsTrafficPageViewWindowBean>() {
            @Override
            public void processElement(String value, ProcessFunction<String, DwsTrafficPageViewWindowBean>.Context ctx, Collector<DwsTrafficPageViewWindowBean> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject page = jsonObject.getJSONObject("page");
                JSONObject common = jsonObject.getJSONObject("common");
                String pageId = page.getString("page_id");
                if ("home".equals(pageId) || "course_list".equals(pageId) || "course_detail".equals(pageId)) {
                    out.collect(DwsTrafficPageViewWindowBean.builder().mid(common.getString("mid")).pageId(pageId).ts(jsonObject.getLong("ts")).build());
                }
            }
        });
        //按照mid进行分组
        KeyedStream<DwsTrafficPageViewWindowBean, String> keyedDS = beanDS.keyBy(DwsTrafficPageViewWindowBean::getMid);
        //使用flink状态编程
        SingleOutputStreamOperator<DwsTrafficPageViewWindowBean> uvBeanDS = keyedDS.process(new KeyedProcessFunction<String, DwsTrafficPageViewWindowBean, DwsTrafficPageViewWindowBean>() {
            ValueState<String> homeState;
            ValueState<String> listState;
            ValueState<String> detailState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> homeLastState = new ValueStateDescriptor<>("homeState", String.class);
                homeLastState.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1L)).build());
                homeState = getRuntimeContext().getState(homeLastState);

                ValueStateDescriptor<String> courseListState = new ValueStateDescriptor<>("courseListState", String.class);
                courseListState.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1L)).build());
                listState = getRuntimeContext().getState(courseListState);

                ValueStateDescriptor<String> courseDetailState = new ValueStateDescriptor<>("courseDetailState", String.class);
                courseDetailState.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1L)).build());
                detailState = getRuntimeContext().getState(courseDetailState);
            }

            @Override
            public void processElement(DwsTrafficPageViewWindowBean bean, KeyedProcessFunction<String, DwsTrafficPageViewWindowBean, DwsTrafficPageViewWindowBean>.Context ctx, Collector<DwsTrafficPageViewWindowBean> out) throws Exception {
                // 首先将度量值先置为0
                bean.setHomeUvCount(0L);
                bean.setListUvCount(0L);
                bean.setDetailUvCount(0L);
                String curDate = DateFormatUtil.tsToDate(bean.getTs());

                String homeLastDt = homeState.value();
                if (homeLastDt == null || homeLastDt.compareTo(curDate) < 0) {
                    //此时为独立访客
                    bean.setHomeUvCount(1L);
                    homeState.update(curDate);
                    out.collect(bean);
                }

                String listLastDt = listState.value();
                if (listLastDt == null || listLastDt.compareTo(curDate) < 0) {
                    bean.setListUvCount(1L);
                    listState.update(curDate);
                    out.collect(bean);
                }

                String detailLastDt = detailState.value();
                if (detailLastDt == null || detailLastDt.compareTo(curDate) < 0) {
                    bean.setDetailUvCount(1L);
                    detailState.update(curDate);
                    out.collect(bean);
                }
            }
        });

        //提取事件时间字段和设置水位线
        SingleOutputStreamOperator<DwsTrafficPageViewWindowBean> withWaterMarkStream = uvBeanDS.assignTimestampsAndWatermarks(WatermarkStrategy.<DwsTrafficPageViewWindowBean>forBoundedOutOfOrderness(Duration.ofSeconds(5L)).withTimestampAssigner(new SerializableTimestampAssigner<DwsTrafficPageViewWindowBean>() {
            @Override
            public long extractTimestamp(DwsTrafficPageViewWindowBean element, long recordTimestamp) {
                return element.getTs();
            }
        }));
        //开窗聚合
        SingleOutputStreamOperator<DwsTrafficPageViewWindowBean> reduceStream = withWaterMarkStream.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10L))).reduce(new ReduceFunction<DwsTrafficPageViewWindowBean>() {
            @Override
            public DwsTrafficPageViewWindowBean reduce(DwsTrafficPageViewWindowBean value1, DwsTrafficPageViewWindowBean value2) throws Exception {
                value1.setHomeUvCount(value1.getHomeUvCount() + value2.getHomeUvCount());
                value1.setListUvCount(value1.getListUvCount() + value2.getListUvCount());
                value1.setDetailUvCount(value1.getDetailUvCount() + value2.getDetailUvCount());
                return value1;
            }
        }, new ProcessAllWindowFunction<DwsTrafficPageViewWindowBean, DwsTrafficPageViewWindowBean, TimeWindow>() {
            @Override
            public void process(Context context, Iterable<DwsTrafficPageViewWindowBean> elements, Collector<DwsTrafficPageViewWindowBean> out) throws Exception {
                // 添加窗口的开始时间和关闭时间
                String stt = DateFormatUtil.tsToDateTime(context.window().getStart());
                String edt = DateFormatUtil.tsToDateTime(context.window().getEnd());
                for (DwsTrafficPageViewWindowBean element : elements) {
                    element.setStt(stt);
                    element.setEdt(edt);
                    element.setTs(System.currentTimeMillis());
                    out.collect(element);
                }

            }
        });
        reduceStream.print("reduceStream");
        //将结果导入到doris中
        reduceStream.map(new BeanToJsonStrMapFunction<>()).sinkTo(FlinkSinkUtil.getDorisSink("dws_traffic_page_view_window"));
    }
}
